package com.atguigu.bigdata.spark.core.rdd.persist

import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.{SparkConf, SparkContext}

/**
  * create by undeRdoG on  2021-06-14  14:07
  * 凡心所向，素履以往，生如逆旅，一苇以航。
  */
object Spark04_RDD_Persist {


  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local").setAppName("WordCount")
    val sc = new SparkContext(sparkConf)

    /**
    *  一般 checkpoint 路径都是在 分布式文件存储系统中，如 HDFS
    * */
    sc.setCheckpointDir("cp")

    val list = List("Hello Scala", "Hello Spark")

    val rdd: RDD[String] = sc.makeRDD(list)

    val flatRDD = rdd.flatMap(_.split(" "))

    val mapRDD = flatRDD.map(x => {
      (x, 1)
    })


    /**
      * checkpoint 一定会落盘  需要指定保存路径
      * 而持久化也可以选择落盘到 磁盘。但是任务运行结束后悔删除该文件，因此不需要制定路径
      **/

    mapRDD.checkpoint()

    val res: RDD[(String, Int)] = mapRDD.reduceByKey(_ + _)

    res.collect().foreach(println)


    val res2: RDD[(String, Iterable[Int])] = mapRDD.groupByKey()

    res2.collect().foreach(println)

  }
}
